package org.niit.service

import java.io.{FileWriter, PrintWriter}
import java.text.SimpleDateFormat

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.niit.common.TService
import org.niit.handler.DataHandler
import org.niit.bean.AdClickData
import org.niit.util.MyKafkaUtil

import scala.collection.mutable.ListBuffer

class TimeCountService extends TService{

  override def dataAnalysis(data:DStream[AdClickData]): Any = {

    /*
      统计一分内广告点击量，每10秒合并一次
      x : 8:33:00  8:33:10 8:33:20 8:33:30 8:33:40 8:33:50 8:34:00
        8:33:00  -- 8:34:00
      在9:06点击了广告 ---> 9:00
         --> 55 --> 50  ==>  55 / 10  =5  5ss*10 = 50    1s = 1000ms  1min = 60s
       获取数据的时候 点击广告时间为时间戳 1680793352123 --> 2023-04-06 23:02:32 --> 2023-04-06 23:02:30
                                    1680793352123 / (1000*10) * 10000 = 1680793350000 (2023-04-06 23:02:30)
                                    1680793352123 / (60 * 10000) * (60 * 10000) = 1680793200000 (2023-04-06 23:00:00)
          48 / 10 = 4
          4 * 10 = 40
     */
    //将点击广告的时间 进行转换
    val reduceDS = data.map(data => {
      val ts = data.ts.toLong // 1680793352123
      val newTs = ts / 10000 * 10000 //1680793350000
      //将转换后的时间作为key value还是点击次数 1
      (newTs, 1)

    }).reduceByKeyAndWindow(
      (x: Int, y: Int) => {
        x + y
      },
      Seconds(60),//统计一分钟内的数据
      Seconds(10)//每10秒统计一次
    )
    /*
      x : 8:33:00  : (8:33:00 - 8:33:09) 15
           8:33:10 : (8:33:10 - 8:33:19) 34
           8:33:20 : (8:33:20 - 8:33:29) 18
           8:33:30 : (8:33:30 - 8:33:39) 60
     */
    reduceDS.foreachRDD(
      rdd=>{
        //新建列表，用来存储最终结果（格式化后的）数据
        val list = ListBuffer[String]()
        //获取的数据，根据key,进行升序排序 并返回一个数组
        val datas = rdd.sortByKey(true).collect()
        //得到数组，进行遍历 并根据模式匹配对数据进行格式化
        datas.foreach{
          case (time,count)=>{
            //1.对时间进行格式化 只保留 时：分：秒
            val sdf = new SimpleDateFormat("HH:mm:ss")
            val timeStr = sdf.format(new java.util.Date(time.toLong))//格式化后的时间--》8:33:20
            // 组装数据，根据 { "xtime":"59:50", "yval":"88" }  来进行组装
            list.append(s"""{"xtime":"${timeStr}", "yval":"${count}"} """ )
          }
        }
        //输出文件--输出到adclick.json
        val out = new PrintWriter(new FileWriter("data/adclick.json") )
        out.println("["+list.mkString(",")+"]" )
        out.flush()
        out.close()
      }
    )


  }
}
